﻿using Confluent.Kafka;
using System;
using System.Diagnostics;
using System.Net.Http;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Xml.Linq;
using System.Collections.Concurrent;
using System.Linq;
using Microsoft.AspNetCore.Http;
using System.Collections.Generic;
using Microsoft.AspNetCore.Http.Internal;
using System.IO;

namespace Message_Check
{
    class Program
    {
        private static readonly HttpClient client = new HttpClient();
        
        private static int numberOfRequests = 3600;// 请求总数
        private static int concurrencyLevel = 1;// 并发数量
        private static int i = 0;
        private static bool checkstatus = true;

        static async Task Maintest(string[] args)
        {
           await KafkaConsume("1010");
        }
        static async Task Main(string[] args)
        {
            Stopwatch stopwatch = new Stopwatch();
            stopwatch.Start();

            int requestCount = 6000;         // 请求总数
            int concurrencyLevel = 100;      // 并发数量

            HttpClient client = new HttpClient();

            ConcurrentBag<long> responseTimes = new ConcurrentBag<long>(); // 记录每个请求的响应时间
            int successCount = 0;
            int failureCount = 0;

            // 启动计时器
            stopwatch.Start();

            // 初始化任务数组
            var tasks = new Task[concurrencyLevel];

            // 并发执行请求
            for (int i = 0; i < requestCount; i += concurrencyLevel)
            {
                int tasksToRun = Math.Min(concurrencyLevel, requestCount - i);

                for (int j = 0; j < tasksToRun; j++)
                {
                    tasks[j] = Task.Run(async () =>
                    {
                        Stopwatch requestWatch = Stopwatch.StartNew();  // 记录单个请求的耗时
                        try
                        {
                            ////接受消息
                            //var respo=KafkaConsume("1010");
                            bool response = await KafkaSendMessage("C_MCD_test");

                            requestWatch.Stop();
                            responseTimes.Add(requestWatch.ElapsedMilliseconds); // 添加响应时间
                            if (response)
                            {
                                successCount++;
                            }
                            else
                            {
                                failureCount++;
                            }
                        }
                        catch (Exception)
                        {
                            requestWatch.Stop();
                            responseTimes.Add(requestWatch.ElapsedMilliseconds); // 添加响应时间
                            failureCount++;
                        }
                    });
                }
                // 等待当前批次的请求完成
                await Task.WhenAll(tasks);
            }

            // 停止计时器
            stopwatch.Stop();

            // 计算平均响应时间和吞吐量
            double averageResponseTime = responseTimes.Count > 0 ? (double)responseTimes.Sum() / responseTimes.Count : 0;
            double throughput = requestCount / (stopwatch.ElapsedMilliseconds / 1000.0);

            // 输出结果
            Console.WriteLine("压测结果:");
            Console.WriteLine($"请求总数: {requestCount}");
            Console.WriteLine($"成功请求: {requestCount- failureCount}");
            Console.WriteLine($"失败请求: {failureCount}");
            Console.WriteLine($"总耗时: {stopwatch.ElapsedMilliseconds} ms");//  Console.WriteLine($"Total time: {stopwatch.Elapsed.TotalSeconds} seconds");
            Console.WriteLine($"平均响应时间: {averageResponseTime} ms");
            Console.WriteLine($"吞吐量 (Requests per Second): {throughput} rps");
        }
        /// <summary>
        /// 发送消息测试
        /// </summary>
        /// <returns></returns>
        public static async Task<bool> KafkaSendMessage(string topics)
        {
            
            string bootstrapServers = "10.159.44.174:9092"; //Kafka集群地址 10.159.44.173:9092,10.159.44.171:9092
            string topic = topics; //topic生产主题 kafak topic: 520、550、620
            var config = new ProducerConfig { BootstrapServers = bootstrapServers };
            string values = "<message><header reply=\"1\"messageVersion=\"2.0\"messageClass=\"511\"transactionID=\"15\"><location routeID=\"1\"routeName=\"NRNPMLine\"equipmentID=\" 1\"equipmentName=\"\"zoneID=\"-1\"zonePos=\"-1\"zoneName=\"\"controllerGuid=\"nrexpress\"laneNo=\"0\"/></header><body><operation operator=\"root\"/><material id=\"16030402c\"programName=\"MutiRefProd\"feederBarcode=\"FA0218AXA842011\"gty=\"-1\"prevReelId=\"\"splicedReelId=\"\"/><offline walue=\"“8”/\"><machineside value=\"F\"/></offline></body></message>";
            string value1 = "<message><header reply=\"1\"messageVersion=\"2.0\"messageClass=\"1002\"transactionID=\"15\"><location routeID=\"1\"routeName=\"NRNPMLine\"equipmentID=\" 1\"equipmentName=\"\"zoneID=\"-1\"zonePos=\"-1\"zoneName=\"\"controllerGuid=\"nrexpress\"laneNo=\"0\"/></header><body><operation operator=\"root\"/><material id=\"16030402c\"programName=\"MutiRefProd\"feederBarcode=\"FA0218AXA842011\"gty=\"-1\"prevReelId=\"\"splicedReelId=\"\"/><offline walue=\"0\"><machineside value=\"F\"/></offline></body></message>";
            using (var producer = new ProducerBuilder<Null, string>(config).Build())
            {
                try
                {
                    //Thread.Sleep(1000);
                    var result = await producer.ProduceAsync(topic, new Message<Null, string>
                    {
                        Value = values,

                    });
                    if (result.Status == Confluent.Kafka.PersistenceStatus.Persisted) //result.Offset.Value 偏移量
                    {
                        checkstatus = true;
                        Console.WriteLine($"succes:第'{i++}'条 状态: '{result.Status}'");
                        Console.WriteLine($"succes:请求体 '{result.Value}'");
                    }
                    else
                    {
                        checkstatus = false;
                        Console.WriteLine($"失败:第'{i++}'条 返回结果: '{result}'");
                    }

                }
                catch (ProduceException<Null, string> e)
                {
                    Console.WriteLine($"Delivery failed: {e.Error.Reason}");
                }
            }
            return checkstatus;
        }

        /// <summary>
        /// 消费（接收消息）
        /// </summary>
        public static async Task KafkaConsume(string topics)
        {
            string bootstrapServers = "10.159.44.174:9092"; //Kafka集群地址
            string topic = topics; //topic消费的主题
            string groupId = "savegen2pana_" + topic; //消费者群组的ID
            try
            {
                var config = new ConsumerConfig
                {
                    BootstrapServers = bootstrapServers,
                    GroupId = groupId,
                    AutoOffsetReset = AutoOffsetReset.Earliest,
                    FetchMaxBytes = 5 * 1024 * 1024, // 5 MB
                    FetchMinBytes = 1 * 1024 * 1024, // 1 MB
                    EnableAutoCommit = false // Disable auto-commit to control commit manually
                };

                using (var consumer = new ConsumerBuilder<Ignore, string>(config).Build())
                {
                    consumer.Subscribe(topic);
                    var cancellationTokenSource = new CancellationTokenSource();
                    var cancellationToken = cancellationTokenSource.Token;
                    try
                    {
                        while (!cancellationToken.IsCancellationRequested)
                        {
                            var consumeResult = consumer.Consume(TimeSpan.FromMilliseconds(100));
                            if (consumeResult != null)
                            {
                                Console.WriteLine($"第{i++}次循环 成功获取数据");
                                Console.WriteLine($"Received message: {consumeResult.Message.Value} at: {consumeResult.TopicPartitionOffset}");
                                consumer.Commit(consumeResult);
                            }
                            else
                            {
                                Console.WriteLine($"第{i++}次循环 无数据");
                                Console.WriteLine("No message received. Doing other work...");
                                if (i > 20) { break; }
                            }

                        }
                    }
                    catch (ConsumeException e)
                    {
                        consumer.Close();
                        Console.WriteLine($"Consume error: {e.Error.Reason}");
                    }

                    finally
                    {
                        consumer.Close();
                    }
                }
            }
            catch (Exception e)
            {
                Console.WriteLine($"Consume error: {e.Message}");
            }
        }

        static async Task SendMessage_new()
        {
            int requestCount = 1000;         // 请求总数
            int concurrencyLevel = 100;      // 并发数量
            string url = "https://your-api-endpoint.com"; // 目标URL
            string postData = "{\"key\": \"value\"}";    // 请求体数据 (JSON 示例)

            HttpClient client = new HttpClient();
            Stopwatch stopwatch = new Stopwatch();
            ConcurrentBag<long> responseTimes = new ConcurrentBag<long>(); // 记录每个请求的响应时间
            int successCount = 0;
            int failureCount = 0;

            // 启动计时器
            stopwatch.Start();

            // 初始化任务数组
            var tasks = new Task[concurrencyLevel];

            // 并发执行请求
            for (int i = 0; i < requestCount; i += concurrencyLevel)
            {
                int tasksToRun = Math.Min(concurrencyLevel, requestCount - i);

                for (int j = 0; j < tasksToRun; j++)
                {
                    tasks[j] = Task.Run(async () =>
                    {
                        Stopwatch requestWatch = Stopwatch.StartNew();  // 记录单个请求的耗时
                        try
                        {
                            StringContent content = new StringContent(postData, Encoding.UTF8, "application/json");
                            HttpResponseMessage response = await client.PostAsync(url, content);
                            requestWatch.Stop();

                            responseTimes.Add(requestWatch.ElapsedMilliseconds); // 添加响应时间

                            if (response.IsSuccessStatusCode)
                            {
                                successCount++;
                            }
                            else
                            {
                                failureCount++;
                            }
                        }
                        catch (Exception)
                        {
                            requestWatch.Stop();
                            responseTimes.Add(requestWatch.ElapsedMilliseconds); // 添加响应时间
                            failureCount++;
                        }
                    });
                }

                // 等待当前批次的请求完成
                await Task.WhenAll(tasks);
            }

            // 停止计时器
            stopwatch.Stop();

            // 计算平均响应时间和吞吐量
            double averageResponseTime = responseTimes.Count > 0 ? (double)responseTimes.Sum() / responseTimes.Count : 0;
            double throughput = requestCount / (stopwatch.ElapsedMilliseconds / 1000.0);

            // 输出结果
            Console.WriteLine("压测结果:");
            Console.WriteLine($"请求总数: {requestCount}");
            Console.WriteLine($"成功请求: {successCount}");
            Console.WriteLine($"失败请求: {failureCount}");
            Console.WriteLine($"总耗时: {stopwatch.ElapsedMilliseconds} ms");
            Console.WriteLine($"平均响应时间: {averageResponseTime} ms");
            Console.WriteLine($"吞吐量 (Requests per Second): {throughput} rps");
        }
        static async Task SendRequestsing()
        {
            for (int i = 0; i < numberOfRequests / concurrencyLevel; i++)
            {
                try
                {
                    var url = "http://localhost:7000/v1/KafkaManagement-KafkaConsume";
                    var jsonContent = "{\"param1\":\"1\"}";
                    var content = new StringContent(jsonContent, Encoding.UTF8, "application/json");
                    //定义 POST 请求的参数

                    var formContent = new MultipartFormDataContent
                   {
                        { new StringContent("value1"), "param1" },
                        { new StringContent("value2"), "param2" },
                           // 如需上传文件，可使用以下方式
                           // { new ByteArrayContent(fileBytes), "file", "filename.ext" }
                       };

                    // 发送 POST 请求
                    var response = await client.PostAsync(url, content);
                    Console.WriteLine($"Response code: {checkstatus}");
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"Request failed: {ex.Message}");
                }

            }
        }
        static async Task SendRequests1()
        {

            var xmlContent = new XDocument(
                                    new XElement("Request",
                                        new XElement("Header",
                                            new XElement("AuthToken", "your_token_here")
                                        ),
                                        new XElement("Body",
                                            new XElement("Param1", "value1"),
                                            new XElement("Param2", "value2")
                                        )
                                    )
                                );
            //var responseContent = await response.Content.ReadAsStringAsync();
            //var xmlResponse = XDocument.Parse(responseContent);
            //var result = xmlResponse.Element("Response").Element("Result").Value;
            //Console.WriteLine($"Result: {result}");

            for (int i = 0; i < numberOfRequests / concurrencyLevel; i++)
            {
                try
                {
                    // 定义 POST 请求的参数
                    var url = "https://your-api-endpoint.com/api/resource";
                    var formContent = new MultipartFormDataContent
                {
                    { new StringContent("value1"), "param1" },
                    { new StringContent("value2"), "param2" },
                    // 如需上传文件，可使用以下方式
                    // { new ByteArrayContent(fileBytes), "file", "filename.ext" }
                };

                    // 发送 POST 请求
                    var response = await client.PostAsync(url, formContent);
                    Console.WriteLine($"Response code: {response.StatusCode}");
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"Request failed: {ex.Message}");
                }
            }
        }
        public class PostFileList
        {
            public string fileName { get; set; }
            public string chunk { get; set; }
            public string chunks { get; set; }
            public string fileType { get; set; }
            public string contain { get; set; }
            public string group { get; set; }
            public string guid { get; set; }
            public IFormCollection formFile { get; set; }
        }

        public string PostCrbFile(IFormCollection formCollection, ref Dictionary<string, PostFileList> postCRBList)
        {
            string exeRes;
            try
            {
                PostFileList CRBfile = null;
                FormFileCollection fileCollection = (FormFileCollection)formCollection.Files;
                string side = string.Empty;
                string kind = string.Empty;
                foreach (var type in formCollection)
                {
                    if (type.Key == "type")
                        side = type.Value.ToString().ToUpper();
                    else if (type.Key == "kind")
                        kind = type.Value.ToString().ToUpper();
                }
                foreach (IFormFile file in fileCollection)
                {
                    CRBfile = new PostFileList();
                    StreamReader reader = new StreamReader(file.OpenReadStream());
                    CRBfile.contain = reader.ReadToEnd();
                    CRBfile.fileName = file.FileName;
                    CRBfile.fileType = Path.GetExtension(file.FileName).ToLower();
                    if (postCRBList.ContainsKey(side))
                        postCRBList[side] = CRBfile;
                    else
                        postCRBList.Add(side, CRBfile);
                }
                exeRes = "OK";
            }
            catch (Exception ex)
            {
                exeRes = "ERROR";
            }
            return exeRes;
        }
    }
}
